package com.example.mapreduce.reduceJoin;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 * ClassName: JoinMapper
 * Package: com.example.mapreduce.reduceJoin
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-17
 * Time: 15:35
 */

//实现mapper 输入的参数依然是一行 偏移量 和 一行内容  输出是pid 和 封装bean对象
public class JoinMapper extends Mapper<LongWritable, Text, Text, TableBean> {

    private Text outK = new Text();

    private TableBean outV = new TableBean();


    private String file;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //初始化方法 就希望得到对应的文件名称
        FileSplit inputSplit = (FileSplit) context.getInputSplit(); //获取切片信息
        //通过切片信息获取问及按摩 默认的切片规则是一个文件 一个切片 一个MapTask
        file = inputSplit.getPath().getName();
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1.获取一行
        String line = value.toString();

        //2.判断是哪个文件
        if (file.contains("order")) {
            //如果包含order说明是order文件
            String[] split = line.split("\t");
            //封装对应的KV
            outK.set(split[1]);
            outV.setId(split[0]);
            outV.setPid(split[1]);
            outV.setAmount(Integer.parseInt(split[2]));
            //没有的属性也要赋值 复为空 不然序列化会出现问题
            outV.setPname("");
            outV.setFlag("order"); //标记的作用 这是一个order表
        } else {
            //一共就两张表 不是order就是pd
            String[] split = line.split("\t");
            outK.set(split[0]);
            outV.setId("");
            outV.setPid(split[0]);
            outV.setAmount(0);
            outV.setPname(split[1]);
            outV.setFlag("pd");
        }
        //写出
        context.write(outK, outV);
    }
}
